package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLogSplitProcessFunction extends ProcessFunction<String, String> {

    private final OutputTag<String> errTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public AppLogSplitProcessFunction(OutputTag<String> errTag, OutputTag<String> startTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errTag = errTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(s);
        if (null != jsonObject.getString("err")){
            context.output(errTag, s);
            jsonObject.remove("err");
        }
        if (null != jsonObject.getString("start")){
            context.output(startTag, s);
        }
        else {
            JSONArray displaysArray = jsonObject.getJSONArray("displays");
            JSONArray actionsArray = jsonObject.getJSONArray("actions");
            jsonObject.remove("displays");
            jsonObject.remove("actions");
            collector.collect(jsonObject.toJSONString());
            if (null != displaysArray){
                for (int i = 0; i < displaysArray.size(); i++) {
                    JSONObject jsonObject1 = displaysArray.getJSONObject(i);
                    jsonObject1.put("display",jsonObject1);
                    context.output(displayTag, jsonObject1.toJSONString());
                }
            }
            if (null != actionsArray){
                for (int i = 0; i < actionsArray.size(); i++) {
                    JSONObject jsonObject1 = actionsArray.getJSONObject(i);
                    jsonObject1.put("action",jsonObject1); // Fixed typo here
                    context.output(actionTag, jsonObject1.toJSONString());
                }
            }
        }
    }
}
